/*
 * Copyright 2012 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */

package io.netty.buffer;

import io.netty.util.internal.ObjectPool.Handle;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.FileChannel;
import java.nio.channels.GatheringByteChannel;
import java.nio.channels.ScatteringByteChannel;

/**
 * 其对象主要由内存池分配器PooledByteBufAllocator创建.
 * 比较常用的实现类有两种:
 * (1)基于堆外直接内存池构建的PooledDirectByteBuf,是Netty在进行I/O的读/写的内存分配的默认方式,堆外直接内存可以减少内存数据拷贝的次数
 * (2) 基于堆内内存池构建的PooledHeapByteBuf.
 *
 * 由于创建PooledByteBuf对象的开销大，而且在高并发情况下，当网络I/O进行读/写时会创建大量的实例。
 * 因此，为了降低系统开销，Netty对Buffer对象进行了池化，缓存了Buffer对象，使对此类型的Buffer可进行重复利用。
 * PooledByteBuf是从内存池中分配出来的Buffer，因此它需要包含内存池的相关信息，
 * 如内存块Chunk、PooledByteBuf在内存块中的位置及其本身所占空间的大小等。
 * @param <T>
 */
abstract class PooledByteBuf<T> extends AbstractReferenceCountedByteBuf {

    //对象重复利用,无须每次创建
    private final Handle<PooledByteBuf<T>> recyclerHandle;

    //一块大的内存区域
    protected PoolChunk<T> chunk;
    //定位到chunk中一块连续内存的指针
    protected long handle;
    //chunk中具体的缓存空间
    protected T memory;
    //偏移量
    protected int offset;
    //长度(ByteBuf中可读字节数)
    protected int length;
    //最大可用长度
    int maxLength;
    //线程缓存
    PoolThreadCache cache;
    //临时ByteBuf
    ByteBuffer tmpNioBuf;
    //内存分配器
    private ByteBufAllocator allocator;

    @SuppressWarnings("unchecked")
    protected PooledByteBuf(Handle<? extends PooledByteBuf<T>> recyclerHandle, int maxCapacity) {
        super(maxCapacity);
        this.recyclerHandle = (Handle<PooledByteBuf<T>>) recyclerHandle;
    }

    void init(PoolChunk<T> chunk, ByteBuffer nioBuffer,
              long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
        init0(chunk, nioBuffer, handle, offset, length, maxLength, cache);
    }

    void initUnpooled(PoolChunk<T> chunk, int length) {
        init0(chunk, null, 0, chunk.offset, length, length, null);
    }

    private void init0(PoolChunk<T> chunk, ByteBuffer nioBuffer,
                       long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
        assert handle >= 0;
        assert chunk != null;

        //大内存块默认为16MB,被分配给多个PooledByteBuf
        this.chunk = chunk;
        //chunk中具体的缓存空间
        memory = chunk.memory;
        //将PooledByteBuf转换成ByteBuffer
        tmpNioBuf = nioBuffer;
        //内存分配器,PooledByteBuf是由Arena的分配器构建的
        allocator = chunk.arena.parent;
        //线程缓存,优先从线程缓存中获取
        this.cache = cache;
        //通过这个指针可以得到PooledByteBuf在chunk这棵二叉树中的具体位置
        this.handle = handle;
        //偏移量
        this.offset = offset;
        //长度:实际数据长度
        this.length = length;
        //写指针不能超过PooledByteBuf的最大可用长度
        this.maxLength = maxLength;
    }

    /**
     * Method must be called before reuse this {@link PooledByteBufAllocator}
     */
    final void reuse(int maxCapacity) {
        maxCapacity(maxCapacity);
        resetRefCnt();
        setIndex0(0, 0);
        discardMarks();
    }

    @Override
    public final int capacity() {
        return length;
    }

    @Override
    public int maxFastWritableBytes() {
        return Math.min(maxLength, maxCapacity()) - writerIndex;
    }

    /**
     * 自动扩容
     * @param newCapacity 新的容量值
     * @return
     */
    @Override
    public final ByteBuf capacity(int newCapacity) {
        //若新的容量值与长度相等，则无须扩容，直接返回即可
        if (newCapacity == length) {
            ensureAccessible();
            return this;
        }
        //检查新的容量值是否大于最大允许容量
        checkNewCapacity(newCapacity);
        //非内存池，在新容量值小于最大长度值的情况下，无须重新分配.只需修改索引和数据长度即可
        if (!chunk.unpooled) {
            // If the request capacity does not require reallocation, just update the length of the memory.
           //新的容量值大于长度值
            //在没有超过 Buffer的最大可用长度值时，只需把长度设为新的容量值即可;若超过了最大可用长度值，则只能重新分配
            if (newCapacity > length) {
                if (newCapacity <= maxLength) {
                    length = newCapacity;
                    return this;
                }
            } else if (newCapacity > maxLength >>> 1 &&
                    (maxLength > 512 || newCapacity > maxLength - 16)) {
                // here newCapacity < length
                //当新容量值小于最大可用长度值时，其读/写索引不能超过新容量值
                length = newCapacity;
                trimIndicesToCapacity(newCapacity);
                return this;
            }
        }

        // Reallocation required.
        //由 Arena 重新分配内存并释放旧的内存空间
        chunk.arena.reallocate(this, newCapacity, true);
        return this;
    }

    @Override
    public final ByteBufAllocator alloc() {
        return allocator;
    }

    @Override
    public final ByteOrder order() {
        return ByteOrder.BIG_ENDIAN;
    }

    @Override
    public final ByteBuf unwrap() {
        return null;
    }

    @Override
    public final ByteBuf retainedDuplicate() {
        return PooledDuplicatedByteBuf.newInstance(this, this, readerIndex(), writerIndex());
    }

    @Override
    public final ByteBuf retainedSlice() {
        final int index = readerIndex();
        return retainedSlice(index, writerIndex() - index);
    }

    @Override
    public final ByteBuf retainedSlice(int index, int length) {
        return PooledSlicedByteBuf.newInstance(this, this, index, length);
    }

    protected final ByteBuffer internalNioBuffer() {
        ByteBuffer tmpNioBuf = this.tmpNioBuf;
        if (tmpNioBuf == null) {
            this.tmpNioBuf = tmpNioBuf = newInternalNioBuffer(memory);
        } else {
            tmpNioBuf.clear();
        }
        return tmpNioBuf;
    }

    protected abstract ByteBuffer newInternalNioBuffer(T memory);

    /**
     * 对象回收,把对象属性清空
     */
    @Override
    protected final void deallocate() {
        if (handle >= 0) {
            final long handle = this.handle;
            this.handle = -1;
            memory = null;
            //释放内存
            chunk.arena.free(chunk, tmpNioBuf, handle, maxLength, cache);
            tmpNioBuf = null;
            chunk = null;
            recycle();
        }
    }


    /**
     * 把PooledByteBuf放回对象池Stack中,以便下次使用
     */
    private void recycle() {
        recyclerHandle.recycle(this);
    }

    protected final int idx(int index) {
        return offset + index;
    }

    final ByteBuffer _internalNioBuffer(int index, int length, boolean duplicate) {
        //根据readIndex 获取偏移量offset
        index = idx(index);
        //当duplicate为true 时，在 memory 中创建共享此缓冲区内容的新的字节缓冲区
        // 当duplicate为false时，先从tmpNioBuf 中获取，当tmpNioBuf为空时
        // 再调用newInternalNioBuffer，此处与memory的类型有关，因此其具体实现由子类完成
        //从memory中复制一份内存对象,两者共享缓冲区,但其位置指针独立维护
        ByteBuffer buffer = duplicate ? newInternalNioBuffer(memory) : internalNioBuffer();
        //设置新的ByteBuffer位置及其最大长度
        buffer.limit(index + length).position(index);
        return buffer;
    }

    /**
     * 从memory中创建一份缓存ByteBuffer,
     * 与memory共享底层数据,但读/写索引独立维护
     * @param index
     * @param length
     * @return
     */
    ByteBuffer duplicateInternalNioBuffer(int index, int length) {
        checkIndex(index, length);
        return _internalNioBuffer(index, length, true);
    }

    @Override
    public final ByteBuffer internalNioBuffer(int index, int length) {
        checkIndex(index, length);
        //只有当tmpNioBuf为空时才创建新的共享缓冲区
        return _internalNioBuffer(index, length, false);
    }

    @Override
    public final int nioBufferCount() {
        return 1;
    }

    @Override
    public final ByteBuffer nioBuffer(int index, int length) {
        return duplicateInternalNioBuffer(index, length).slice();
    }

    @Override
    public final ByteBuffer[] nioBuffers(int index, int length) {
        return new ByteBuffer[] { nioBuffer(index, length) };
    }

    @Override
    public final boolean isContiguous() {
        return true;
    }

    /**
     * channel 从poolByteBuf中获取数据,PooledByteBuf的读索引的变化,
     * 由父类AbstractByteBuf的readBytes()方法维护
     * @param index
     * @param out
     * @param length the maximum number of bytes to transfer
     *
     * @return
     * @throws IOException
     */
    @Override
    public final int getBytes(int index, GatheringByteChannel out, int length) throws IOException {
        return out.write(duplicateInternalNioBuffer(index, length));
    }

    @Override
    public final int readBytes(GatheringByteChannel out, int length) throws IOException {
        checkReadableBytes(length);
        int readBytes = out.write(_internalNioBuffer(readerIndex, length, false));
        readerIndex += readBytes;
        return readBytes;
    }

    @Override
    public final int getBytes(int index, FileChannel out, long position, int length) throws IOException {
        return out.write(duplicateInternalNioBuffer(index, length), position);
    }

    @Override
    public final int readBytes(FileChannel out, long position, int length) throws IOException {
        checkReadableBytes(length);
        int readBytes = out.write(_internalNioBuffer(readerIndex, length, false), position);
        readerIndex += readBytes;
        return readBytes;
    }

    @Override
    public final int setBytes(int index, ScatteringByteChannel in, int length) throws IOException {
        try {
            return in.read(internalNioBuffer(index, length));
        } catch (ClosedChannelException ignored) {
            //客户端主动关闭链接,返回-1,触发对应的用户事件
            return -1;
        }
    }

    @Override
    public final int setBytes(int index, FileChannel in, long position, int length) throws IOException {
        try {
            return in.read(internalNioBuffer(index, length), position);
        } catch (ClosedChannelException ignored) {
            return -1;
        }
    }
}
